package com.dstz.sys.simplemq.consumer;

import com.dstz.sys.api.jms.JmsHandler;
import com.dstz.sys.api.jms.constants.JmsDestinationConstant;
import com.dstz.sys.api.jms.model.JmsDTO;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.concurrent.TimeUnit;

/**
 * Redis消息队列消费
 *
 * @author wacxhs
 */
@SuppressWarnings("ALL")
public class RedisMessageQueueConsumer extends AbstractMessageQueue implements DisposableBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisMessageQueueConsumer.class);

    /**
     * reidsTemplate 容器中名称
     */
    private String redisTemplateBeanName;

    /**
     * 处理消息核心线程数
     */
    private int listenThreadSize = 1;

    /**
     * 监听消息队列间隔(ms)
     */
    private long listenInterval = 1000;

    /**
     * 消息队列
     */
    private BoundListOperations<String, Object> messageQueue;

    private ThreadGroup threadGroup;

    public void setRedisTemplateBeanName(String redisTemplateBeanName) {
        this.redisTemplateBeanName = redisTemplateBeanName;
    }

    /**
     * 设置监听队列间隔
     *
     * @param listenInterval 监听间隔(ms)
     */
    public void setListenInterval(long listenInterval) {
        this.listenInterval = listenInterval;
    }

    /**
     * 设置监听线程数
     *
     * @param listenThreadSize 监听线程数
     */
    public void setListenThreadSize(int listenThreadSize) {
        this.listenThreadSize = listenThreadSize;
    }

    @Override
    public void destroy() throws Exception {
        LOGGER.debug("准备关闭Redis消息队列消息消费端");
        this.threadGroup.interrupt();
        LOGGER.debug("关闭Redis消息队列消息消费完毕");
    }

    @SuppressWarnings("unchecked")
    @Override
    protected void containerInitialCompleteAfter() {
        LOGGER.debug("初始化Redis消息队列处理");
        if (StringUtils.isEmpty(this.redisTemplateBeanName)) {
            this.messageQueue = getApplicationContext().getBean(RedisTemplate.class).boundListOps(JmsDestinationConstant.DEFAULT_NAME);
        } else {
            this.messageQueue = getApplicationContext().getBean(redisTemplateBeanName, RedisTemplate.class).boundListOps(JmsDestinationConstant.DEFAULT_NAME);
        }
        RedisQueueListener redisQueueListener = new RedisQueueListener();
        ThreadGroup threadGroup = new ThreadGroup(String.format("redis-queue-%s-listen", JmsDestinationConstant.DEFAULT_NAME));
        threadGroup.setDaemon(true);
        for (int i = 0; i < listenThreadSize; i++) {
            new Thread(threadGroup, redisQueueListener).start();
        }
        this.threadGroup = threadGroup;
    }


    /**
     * Redis队列监听器
     */
    private class RedisQueueListener implements Runnable {

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                LOGGER.trace("监听Redis消息队列({})返回结果......", JmsDestinationConstant.DEFAULT_NAME);
                Object jmsDTO;
                try {
                    jmsDTO = messageQueue.leftPop(listenInterval, TimeUnit.MILLISECONDS);
                } catch (Exception e) {
                    LOGGER.warn("监听Redis消息队列({})返回结果出错", JmsDestinationConstant.DEFAULT_NAME, e);
                    continue;
                }
                if (jmsDTO == null) {
                    continue;
                } else if (!(jmsDTO instanceof JmsDTO)) {
                    LOGGER.warn("Redis消息队列({})返回数据类型({})非[com.dstz.sys.api.jms.model.JmsDTO]", JmsDestinationConstant.DEFAULT_NAME, jmsDTO.getClass());
                    continue;
                }
                JmsDTO data = (JmsDTO) jmsDTO;
                JmsHandler jmsHandler = getJmsHandler(data.getType());
                jmsHandler.handlerMessage(data);
            }
            LOGGER.info("退出Redis消息队列({})监听", JmsDestinationConstant.DEFAULT_NAME);
        }
    }

}
